/******************************************************************
 * Copyright (c) 2024 Bocloud Technologies Co., Ltd.
 * installer is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain n copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 ******************************************************************/

package phases

import (
	"context"
	"strings"
	"sync"
	"time"

	"github.com/pkg/errors"
	confv1beta1 "gopkg.openfuyao.cn/bkecommon/cluster/api/v1beta1"
	bkenode "gopkg.openfuyao.cn/bkecommon/cluster/node"
	"k8s.io/apimachinery/pkg/util/wait"
	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
	"sigs.k8s.io/cluster-api/util/conditions"
	ctrl "sigs.k8s.io/controller-runtime"

	bkev1beta1 "gopkg.openfuyao.cn/cluster-api-provider-bke/api/v1beta1"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/pkg/mergecluster"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/pkg/phaseframe"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/pkg/phaseframe/phaseutil"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils/clusterutil"
	"gopkg.openfuyao.cn/cluster-api-provider-bke/utils/constant"
)

const (
	EnsureWorkerJoinName confv1beta1.BKEClusterPhase = "EnsureWorkerJoin"
)

type EnsureWorkerJoin struct {
	phaseframe.BasePhase
	nodesToJoin bkenode.Nodes
}

func NewEnsureWorkerJoin(ctx *phaseframe.PhaseContext) phaseframe.Phase {
	base := phaseframe.NewBasePhase(ctx, EnsureWorkerJoinName)
	return &EnsureWorkerJoin{BasePhase: base}
}

func (e *EnsureWorkerJoin) Execute() (_ ctrl.Result, err error) {
	if err = e.reconcileWorkerJoin(); err != nil {
		return ctrl.Result{}, err
	}
	return ctrl.Result{}, nil
}

func (e *EnsureWorkerJoin) NeedExecute(old *bkev1beta1.BKECluster, new *bkev1beta1.BKECluster) (needExecute bool) {
	if !e.BasePhase.DefaultNeedExecute(old, new) {
		return false
	}

	// cluster没有初始化的condition，不需要执行
	if err := e.Ctx.RefreshCtxCluster(); err == nil {
		if !conditions.IsTrue(e.Ctx.Cluster, clusterv1.ControlPlaneInitializedCondition) {
			return false
		}
	}

	nodes := phaseutil.GetNeedJoinWorkerNodes(new)
	if nodes.Length() == 0 {
		return false
	}

	e.SetStatus(bkev1beta1.PhaseWaiting)
	return true
}

func (e *EnsureWorkerJoin) getExceptJoinNodes() bkenode.Nodes {
	_, _, bkeCluster, _, log := e.Ctx.Untie()
	nodes := phaseutil.GetNeedJoinWorkerNodes(bkeCluster)
	exceptJoinNodes := bkenode.Nodes{}
	for _, node := range nodes {
		// 检查是否需要跳过该节点
		if bkeCluster.GetNodeStateNeedSkip(node.IP) {
			log.Info(constant.WorkerNodeSkipReason, "Node is marked as skip, skip join node. node: %v", phaseutil.NodeInfo(node))
			continue
		}
		if !bkeCluster.GetNodeStateFlag(node.IP, bkev1beta1.NodeEnvFlag) || !bkeCluster.GetNodeStateFlag(node.IP, bkev1beta1.NodeAgentReadyFlag) {
			continue
		}
		exceptJoinNodes = append(exceptJoinNodes, node)
	}
	return exceptJoinNodes
}

func (e *EnsureWorkerJoin) reconcileWorkerJoin() error {

	ctx, c, bkeCluster, _, log := e.Ctx.Untie()

	if conditions.IsFalse(e.Ctx.Cluster, clusterv1.ControlPlaneInitializedCondition) {
		log.Warn(constant.MasterNotInitReason, "master is not initialized, skip join worker nodes process")
		return nil
	}

	exceptJoinNodes := e.getExceptJoinNodes()
	if exceptJoinNodes.Length() == 0 {
		return nil
	}

	bkeNodes := bkenode.Nodes(bkeCluster.Spec.ClusterConfig.Nodes)

	log.Info(constant.WorkerJoiningReason, "Start join worker nodes process")
	log.Info(constant.WorkerJoiningReason, "Check whether the node has been associated with a Machine to avoid duplicate creation")
	// step 1 get machine ,to avoid duplicate creation
	nodesInfos, nodesCount, err2 := e.getJoinableNodesInfo(exceptJoinNodes)
	if err2 != nil {
		return err2
	}

	if clusterutil.IsBocloudCluster(bkeCluster) {
		log.Info(constant.WorkerJoiningReason, "bocloud cluster need to distribute kube-proxy kubeconfig")
		if err := phaseutil.DistributeKubeProxyKubeConfig(ctx, c, bkeCluster, e.nodesToJoin, log); err != nil {
			return err
		}
	}

	log.Info(constant.WorkerJoiningReason, "%d nodes will be joined, nodes: %v", nodesCount, strings.Join(nodesInfos, ", "))
	scope, err := phaseutil.GetClusterAPIAssociateObjs(ctx, c, e.Ctx.Cluster)
	if err != nil || scope.MachineDeployment == nil {
		log.Error(constant.WorkerJoinFailedReason, "Get cluster-api associate objs failed. err: %v", err)
		// cluster api object error, no need to continue
		return err
	}

	specCopy := scope.MachineDeployment.Spec.DeepCopy()
	currentReplicas := specCopy.Replicas
	// 如果节点加入过程中出现异常，需要将节点数量恢复到加入前的状态
	defer func() {
		if err != nil {
			log.Info(constant.WorkerJoinFailedReason, "Scale down MachineDeployment replicas to %d.", *currentReplicas)
			scope.MachineDeployment.Spec.Replicas = currentReplicas
			if err = phaseutil.ResumeClusterAPIObj(ctx, c, scope.MachineDeployment); err != nil {
				log.Error(constant.WorkerJoinFailedReason, "Rollback MachineDeployment replicas failed. err: %v", err)
			}
		}
	}()

	workerNodes := bkeNodes.Worker()

	exceptReplicas := *currentReplicas + int32(nodesCount)
	// 无论如何不能超过bkecluster的worker数量
	if exceptReplicas > int32(workerNodes.Length()) {
		exceptReplicas = int32(workerNodes.Length())
	}

	scope.MachineDeployment.Spec.Replicas = &exceptReplicas

	log.Info(constant.WorkerJoiningReason, "Scale up MachineDeployment replicas %d to %d", *currentReplicas, exceptReplicas)

	if err = phaseutil.ResumeClusterAPIObj(ctx, c, scope.MachineDeployment); err != nil {
		log.Error(constant.WorkerJoinFailedReason, "Scale up MachineDeployment replicas failed. err: %v", err)
		// cluster api object error, no need to continue
		return err
	}
	// wait for machine deployment scale up
	if err = e.waitWorkerJoin(); err != nil {
		log.Error(constant.WorkerJoinFailedReason, "Wait for worker join failed. err: %v", err)
		return err
	}
	return nil
}

func (e *EnsureWorkerJoin) getJoinableNodesInfo(exceptJoinNodes bkenode.Nodes) ([]string, int, error) {
	ctx, c, bkeCluster, _, log := e.Ctx.Untie()

	var nodesInfos []string
	nodesCount := 0
	for _, node := range exceptJoinNodes {
		// 正常应该是找不到关联的machine的
		if _, err := phaseutil.NodeToMachine(ctx, c, bkeCluster, node); err == nil {
			log.Warn(constant.WorkerJoinedReason, "Node already exists, skip join node. node: %v", node.Hostname)
			bkeCluster.MarkNodeStateFlag(node.IP, bkev1beta1.NodeBootFlag)
		} else {
			nodesInfos = append(nodesInfos, phaseutil.NodeInfo(node))
			nodesCount++
			e.nodesToJoin = append(e.nodesToJoin, node)
		}
	}
	if nodesCount == 0 {
		log.Info(constant.WorkerJoinedReason, "All nodes have been joined, skip join node.")
		if err := mergecluster.SyncStatusUntilComplete(c, bkeCluster); err != nil {
			return nil, 0, err
		}
		return nil, 0, nil
	}
	return nodesInfos, nodesCount, nil
}

func (e *EnsureWorkerJoin) waitWorkerJoin() error {
	if len(e.nodesToJoin) == 0 || e.nodesToJoin == nil {
		return nil
	}
	ctx, c, _, _, log := e.Ctx.Untie()

	// 获取超时设置
	timeOut, err := phaseutil.GetBootTimeOut(e.Ctx.BKECluster)
	if err != nil {
		log.Warn(constant.MasterNotInitReason, "Get boot timeout failed. err: %v", err)
	}

	ctxTimeout, cancel := context.WithTimeout(ctx, timeOut)
	defer cancel()

	// 调用轮询函数
	successJoinNode, err := e.pollWorkerJoinStatus(ctxTimeout)
	if err != nil {
		log.Warn("pollWorkerJoinStatus", "pollWorkerJoinStatus failed. err: %v", err)
	}

	// get failed node
	failedNodes := bkenode.Nodes{}
	successNodes := bkenode.Nodes{}
	var success []string
	var failed []string
	for i, node := range e.nodesToJoin {
		if _, ok := successJoinNode.Load(i); !ok {
			failedNodes = append(failedNodes, node)
			failed = append(failed, phaseutil.NodeInfo(node))
		} else {
			successNodes = append(successNodes, node)
			success = append(success, phaseutil.NodeInfo(node))
		}
	}

	// 修补status的node信息
	if successNodes.Length() > 0 {
		if err := e.Ctx.RefreshCtxBKECluster(); err != nil {
			return err
		}
		for _, node := range successNodes {
			e.Ctx.BKECluster.SetNodeStateWithMessage(node.IP, bkev1beta1.NodeNotReady, "Join worker nodes success")
		}
		if err := mergecluster.SyncStatusUntilComplete(c, e.Ctx.BKECluster); err != nil {
			return err
		}
	}

	if errors.Is(err, wait.ErrWaitTimeout) && len(failed) > 0 {
		// 设置失败节点的NeedSkip为true
		for _, node := range failedNodes {
			for _, nodeState := range e.Ctx.BKECluster.Status.NodesStatus {
				if nodeState.Node.IP == node.IP {
					e.Ctx.BKECluster.SetNodeStateNeedSkip(node.IP, true)
					break
				}
			}
		}
		log.Warn(constant.WorkerNodeSkipReason, "Time out wait worker join failed, failed nodes: %d", len(failed))
		for _, node := range failed {
			log.Warn(constant.WorkerNodeSkipReason, "failed node: %v", node)
		}
	}

	if err != nil {
		return err
	}
	log.Info(constant.WorkerJoinSucceedReason, "Worker nodes join success")
	return nil
}

func (e *EnsureWorkerJoin) pollWorkerJoinStatus(ctxTimeout context.Context) (*sync.Map, error) {
	ctx, c, bkeCluster, _, log := e.Ctx.Untie()
	successJoinNode := sync.Map{}

	pollCount := 0
	err := wait.PollImmediateUntil(1*time.Second, func() (done bool, err error) {
		pollCount++
		wg := sync.WaitGroup{}
		for i, node := range e.nodesToJoin {
			wg.Add(1) //在启动goroutine前执行Add, 确保时机正确
			go func(index int, n confv1beta1.Node) {
				defer wg.Done()
				if _, ok := successJoinNode.Load(index); ok {
					return
				}
				machine, err := phaseutil.NodeToMachine(ctx, c, bkeCluster, n)
				if err != nil {
					return
				}
				if machine.Status.NodeRef != nil {
					successJoinNode.Store(index, n)
				}
			}(i, node)
		}
		wg.Wait()

		success := 0
		successJoinNode.Range(func(key, value interface{}) bool { success++; return true })
		if success != len(e.nodesToJoin) {
			const logFrequency = 10
			if pollCount%logFrequency == 0 {
				log.Info(constant.WorkerJoiningReason, "Wait worker join. success: %d, total: %d", success, len(e.nodesToJoin))
			}
			return false, nil
		}
		log.Info(constant.WorkerJoiningReason, "Wait worker join. success: %d, total: %d", success, len(e.nodesToJoin))

		return true, nil
	}, ctxTimeout.Done())
	return &successJoinNode, err
}
